package top.wangjiu.reactive.rpc.handler;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.reactivestreams.Subscription;
import reactor.core.publisher.Flux;
import top.wangjiu.reactive.rpc.model.RpcRequest;
import top.wangjiu.reactive.rpc.model.RpcResponse;
import top.wangjiu.reactive.rpc.server.ServerHander;

/**
 * @ClassName MonoResultTypeHandler
 * @Description TODO
 * @Author wangjiu
 * @Date 2021/8/3 5:03 下午
 */
public class FluxRequestHandler extends AbstractRequestHandler<Flux> {

    private static final Log log = LogFactory.getLog(FluxRequestHandler.class);

    private volatile Flux result;

    private volatile String requestId;

    private volatile Subscription subscription;

    @Override
    public void onRequest(RpcRequest request) {
        if(requestId == null){
            requestId = request.getRequestId();

            try {
                result = invoke(request);
                result.doOnComplete(()->{
                    RpcResponse response = new RpcResponse(requestId);
                    response.setLast(true);
                    write(response);
                }).doOnError(e->{
                    RpcResponse response = new RpcResponse(requestId);
                    response.setLast(true);
                    response.setSuccess(false);
                    response.setCause((Throwable) e);
                    write(response);
                }).subscribe(r->{
                    RpcResponse response = new RpcResponse(requestId);
                    response.setResult(r);
                    write(response);
                },null,null,s -> {
                    subscription = ((Subscription) s);
                    if(request.isCancel()){
                        subscription.cancel();
                    }else {
                        subscription.request(request.getCount());
                    }
                });
            } catch (Exception e) {
                log.error("invoke method error:"+request,e);
                RpcResponse response = new RpcResponse(requestId);
                response.setLast(true);
                response.setSuccess(false);
                response.setCause(e);
                write(response);
            }
        }else if(subscription != null){
            subscription.request(request.getCount());
            if(request.isCancel()){
                subscription.cancel();
            }
        }else{

        }
    }

    @Override
    public void onDisconnect() {
        if(subscription != null){
            subscription.cancel();
        }
    }
}
